﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Net.Sockets;
using System.Reflection;
using MapReduce.Services;
using System.Threading;
using MapReduce.Comm.IO; 

namespace MapReduce.Comm
{
    public class SocketThread : BaseComm
    {
        //private String mAddress;
        //private ushort mPort;

        public SocketThread(Socket pSocket, TcpListener pTcpListener, String pAddress, ushort pPort)
        {
            //String []endPt = pTcpListener.LocalEndpoint.ToString().Split(new char[] { ':' });
            //mAddress = endPt[0];
            //mPort = ushort.Parse(endPt[1]);

            this.IOStream = new SocketIOStream(pSocket);
        }//End SocketThread


        public void Start()
        {
            Thread t = new Thread(ThreadedStart);
            t.Start();
        }
        //Everything is in following format
        // Byte 0-99: 
        //      Request Name
        // Byte 100-107 (inclusive):
        //      Unsigned long value. Length of incoming data.
        // Byte 108-... : Actual Data

        public void ThreadedStart()
        {
            byte[][] dataList =null;
            String className = null;
            String callbackId = null;
            try
            {

                byte[] b = this.ReceiveData();

                StringBuilder serviceName = U.Convert(b);
                className = serviceName.ToString().Trim();

                //Get the callback id 'if response is expected
                callbackId = U.Convert(this.ReceiveData()).ToString();

                U.Log("Executing : " + serviceName + " callbackId:" + callbackId);



                //Get length
                b = this.ReceiveData();
                ushort totalDataArrays = BitConverter.ToUInt16(b, 0);

                dataList = new byte[totalDataArrays][];
                for (int i = 0; i < totalDataArrays; i++)
                {
                    dataList[i] = this.ReceiveData();
                }

            }
            finally
            {
                //Clean up
                this.Close();
            }

            byte[][] opData = null;
            byte callStatus = JobRequest.SUCCESS;
            try
            {
                //After closing the request stream, start processing the service
                System.Type classType = Assembly.GetExecutingAssembly().GetType(className, true);
                IService service = (IService)System.Activator.CreateInstance(classType);
                opData = service.Execute(dataList);

            }
            catch (Exception ex)
            {
                String msg = ex.ToString();
                int i= msg.IndexOf("at MapReduce.Services.");
                if (i > 0)
                {
                    msg = msg.Substring(0, i);
                }
                opData = new byte[][] { U.GetBytes(msg)};
                callStatus = JobRequest.ERROR;
            }

            if (callbackId != "0")
            {
                //Find out what port job requester is listening on
                //When requester expects callback, the callbackId has form callbackAddress:Port:guid
                String[] callbackArr = callbackId.Split(new char[] { ':' });

                List<byte[]> newOpList = new List<byte[]>();

                newOpList.Add(U.GetBytes(callbackArr[2])); //callback guid.
                newOpList.Add(new byte[]{callStatus}); //error or success? 1:success 0:error

                //append real output data
                foreach (byte[] barr in opData)
                {
                    newOpList.Add(barr);
                }

                //send this output data back to the requester
                JobRequest req = new JobRequest(callbackArr[0], ushort.Parse(callbackArr[1]));
                req.Send("MapReduce.Services.OutputService", newOpList.ToArray(), null);
            }

        }//ThreadedStart



    }
}
